package com.danan.realtime.app.dwd;

import com.danan.realtime.app.BaseSqlApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.util.SinkUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/1/25
 */
public class DwdApp_3_trade_order_detail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdApp_3_trade_order_detail().init(4003,3,"DwdApp_3_trade_order_detail");
    }
    @Override
    public void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 读取ods_db的数据
        readOdsDb(tableEnv,"DwdApp_3_trade_order_detail");
        // 2 过滤订单相关的数据
        filterOrderDetail(tableEnv);
        // 3 读取字典表数据
        readBaseDic(tableEnv);
        // 4 join各表，并进行维度退化
        joinAndExtendDim(tableEnv);
        // 5 写出结果到Kafka（upsertKafka）
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建目标表，与Kafka主题关联
        tableEnv.executeSql("create table target (" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "sku_id string," +
                "sku_name string," +
                "province_id string," +
                "activity_id string," +
                "activity_rule_id string," +
                "coupon_id string," +
                "date_id string," +
                "create_time string," +
                "source_id string," +
                "source_type string," +
                "source_type_name string," +
                "sku_num string," +
                "split_original_amount string," +
                "split_activity_amount string," +
                "split_coupon_amount string," +
                "split_total_amount string," +
                "ts bigint ," +
                "primary key (id) not enforced " +
                ")" +
                SinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        // 数据写入
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinAndExtendDim(StreamTableEnvironment tableEnv) {
        // 设置数据的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        // join
        Table joinResult = tableEnv.sqlQuery("select " +
                "od.id," +
                "od.order_id," +
                "oi.user_id," +
                "od.sku_id," +
                "od.sku_name," +
                "oi.province_id," +
                "act.activity_id," +
                "act.activity_rule_id," +
                "cou.coupon_id," +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id," +
                "od.create_time," +
                "od.source_id," +
                "od.source_type," +
                "dic.dic_name source_type_name," +
                "od.sku_num," +
                "od.split_original_amount," +
                "od.split_activity_amount," +
                "od.split_coupon_amount," +
                "od.split_total_amount," +
                "od.ts " +
                "from order_detail od " +
                "left join order_info oi on od.order_id = oi.id " +
                "left join order_detail_activity act on od.id = act.order_detail_id " +
                "left join order_detail_coupon cou on od.id = cou.order_detail_id " +
                "join base_dic for system_time as of od.pt as dic on od.source_type = dic.dic_code");
        tableEnv.createTemporaryView("join_result",joinResult);

    }

    private void filterOrderDetail(StreamTableEnvironment tableEnv) {
        // order_detail
        Table orderDetail = tableEnv.sqlQuery("select " +
                "`data`['id'] id," +
                "`data`['order_id'] order_id," +
                "`data`['sku_id'] sku_id," +
                "`data`['sku_name'] sku_name," +
                "`data`['create_time'] create_time," +
                "`data`['source_id'] source_id," +
                "`data`['source_type'] source_type," +
                "`data`['sku_num'] sku_num," +
                // 计算原始金额
                "cast(cast(`data`['sku_num'] as decimal(16,2)) * " +
                "       cast(`data`['order_price'] as decimal(16,2)) as String) split_original_amount," +
                "`data`['split_total_amount'] split_total_amount," +
                "`data`['split_activity_amount'] split_activity_amount," +
                "`data`['split_coupon_amount'] split_coupon_amount," +
                "ts," +
                "pt " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_detail' and `type`='insert'");
        tableEnv.createTemporaryView("order_detail",orderDetail);
        // order_info
        Table orderInfo = tableEnv.sqlQuery("select " +
                "`data`['id'] id," +
                "`data`['user_id'] user_id," +
                "`data`['province_id'] province_id " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_info' and `type`='insert'");
        tableEnv.createTemporaryView("order_info",orderInfo);
        // order_detail_activity
        Table orderDetailActivity = tableEnv.sqlQuery("select " +
                "data['order_detail_id'] order_detail_id," +
                "data['activity_id'] activity_id," +
                "data['activity_rule_id'] activity_rule_id " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_detail_activity' and `type`='insert'");
        tableEnv.createTemporaryView("order_detail_activity",orderDetailActivity);
        // order_detail_coupon
        Table orderDetailCoupon = tableEnv.sqlQuery("select " +
                "data['order_detail_id'] order_detail_id," +
                "data['coupon_id'] coupon_id " +
                "from source " +
                "where `database`='gmall2023' and `table`='order_detail_coupon' and `type`='insert'");
        tableEnv.createTemporaryView("order_detail_coupon",orderDetailCoupon);
    }
}
